package com.example.bigdata.kafka.consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.WakeupException;

import java.time.Duration;
import java.util.Collections;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.atomic.AtomicBoolean;

public class ConsumerThreadSample {
    public final static String TOPIC_NAME = "hrsjw1_topic";
    /**
     *  这种类型是经典模式，每一个线程单独创建一个KafkaConsumer 用于保证线程安全
     *  consumer 是阻塞形式处理数据
     */
    public static void main(String[] args) throws InterruptedException {
        KafkaConsumerRunner r1 = new KafkaConsumerRunner();
        Thread t1 = new Thread(r1);
        t1.start();
        Thread.sleep(15000);
        r1.shutdown();
    }
    public static class KafkaConsumerRunner implements Runnable {
        private final AtomicBoolean closed = new AtomicBoolean(false);
        // 在创建 KafkaConsumerRunner 时候 也创建一个 KafkaConsumer 这样来解决线程安全问题
        private final KafkaConsumer consumer;
        // 订阅过程
        public KafkaConsumerRunner(){
            Properties props = new Properties();
            props.setProperty("bootstrap.servers", "localhost:9092");
            props.setProperty("group.id", "test");
            props.setProperty("enable.auto.commit", "false");
            props.setProperty("auto.commit.interval.ms", "1000");
            props.setProperty("session.timeout.ms", "30000");
            props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            consumer = new KafkaConsumer(props);
            TopicPartition p0 = new TopicPartition(TOPIC_NAME, 0);
            TopicPartition p1 = new TopicPartition(TOPIC_NAME, 1);

            consumer.assign(java.util.Arrays.asList(p0,p1));
        }
        public void run(){
            try {
                while (!closed.get()){
                    // 处理消息
                    ConsumerRecords<String,String> records = consumer.poll(Duration.ofMillis(10000));
                    for (TopicPartition partition:records.partitions()){
                        List<ConsumerRecord<String, String>> pRecord = records.records(partition);
                        // 处理每个分区的消息
                        for(ConsumerRecord<String, String> record:pRecord ){
                            System.out.printf("offset = %d, patition = %d ,key = %s, value = %s%n",
                                    record.offset(), record.partition(), record.key(), record.value());
                        }
                        // 返回新offset 给kafka
                        long lastOffset = pRecord.get(pRecord.size()-1).offset();
                        // 注意 +1
                        consumer.commitSync(Collections.singletonMap(partition,new OffsetAndMetadata(lastOffset+1)));
                    }
                }
            }catch (WakeupException e){
                if(!closed.get()){
                    throw e;
                }
            }finally {
                consumer.close();
            }
        }
        public void shutdown(){
            closed.set(true);
            consumer.wakeup();
        }
    }
}
